package yunjiao.javatutorials.guava.concurrent.interrupt;

import java.util.LinkedList;
import java.util.Queue;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;

/**
 * 响应式任务处理器
 *
 * @author yangyunjiao
 */
public class ResponsiveTaskProcessor {
    private final Lock lock = new ReentrantLock();
    private final InterruptAwareCondition taskCondition;
    private final Queue<Runnable> taskQueue = new LinkedList<>();
    private volatile boolean running = true;
    private final Thread processorThread;

    public ResponsiveTaskProcessor() {
        this.taskCondition = new InterruptAwareCondition(lock.newCondition());
        this.processorThread = new Thread(this::processTasks, "TaskProcessor");
    }

    public void start() {
        running = true;
        processorThread.start();
        System.out.println("任务处理器已启动");
    }

    public void submitTask(Runnable task) {
        lock.lock();
        try {
            taskQueue.offer(task);
            taskCondition.signal();
            System.out.println("任务已提交，队列大小: " + taskQueue.size());
        } finally {
            lock.unlock();
        }
    }

    public void stop() {
        System.out.println("停止任务处理器...");
        running = false;

        // 首先尝试优雅停止
        lock.lock();
        try {
            taskCondition.signal();
        } finally {
            lock.unlock();
        }

        try {
            processorThread.join(2000); // 等待 2 秒

            if (processorThread.isAlive()) {
                System.out.println("处理器未及时响应，进行中断");
                processorThread.interrupt();
                processorThread.join(1000);
            }

        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
            System.out.println("停止过程被中断");
        }

        // 分析中断情况
        taskCondition.printInterruptStatistics();
        System.out.println("任务处理器已停止");
    }

    public void stopImmediately() {
        System.out.println("立即停止任务处理器");
        running = false;
        processorThread.interrupt();

        try {
            processorThread.join(1000);
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
        }

        taskCondition.printInterruptStatistics();
    }

    private void processTasks() {
        System.out.println("任务处理线程开始运行");

        while (running && !Thread.currentThread().isInterrupted()) {
            Runnable task = null;

            lock.lock();
            try {
                while (running && taskQueue.isEmpty() && !Thread.currentThread().isInterrupted()) {
                    taskCondition.await(); // 等待新任务
                }

                if (!running) {
                    break;
                }

                if (taskCondition.wasInterrupted()) {
                    System.out.println("检测到等待被中断，检查是否继续运行");
                    // 清除标志以便下次使用
                    taskCondition.clearInterruptFlag();
                }

                task = taskQueue.poll();
            } catch (InterruptedException e) {
                System.out.println("任务处理线程被中断");
                break;
            } finally {
                lock.unlock();
            }

            if (task != null) {
                try {
                    System.out.println("执行任务...");
                    task.run();
                    System.out.println("任务执行完成");
                } catch (Exception e) {
                    System.err.println("任务执行失败: " + e.getMessage());
                }
            }
        }

        System.out.println("任务处理线程结束运行");
        // 处理剩余任务
        processRemainingTasks();
    }

    private void processRemainingTasks() {
        System.out.println("处理剩余任务，队列大小: " + taskQueue.size());
        while (!taskQueue.isEmpty()) {
            Runnable task = taskQueue.poll();
            if (task != null) {
                try {
                    task.run();
                } catch (Exception e) {
                    System.err.println("剩余任务执行失败: " + e.getMessage());
                }
            }
        }
    }

    public static void main(String[] args) throws InterruptedException {
        ResponsiveTaskProcessor processor = new ResponsiveTaskProcessor();
        processor.start();

        // 提交一些任务
        for (int i = 1; i <= 5; i++) {
            final int taskId = i;
            processor.submitTask(() -> {
                System.out.println("执行任务 " + taskId);
                try {
                    Thread.sleep(500);
                } catch (InterruptedException e) {
                    Thread.currentThread().interrupt();
                }
            });
        }

        // 让处理器运行一会儿
        Thread.sleep(2000);

        // 优雅停止
        processor.stop();

        System.out.println("=== 测试立即停止 ===");

        // 测试立即停止
        ResponsiveTaskProcessor processor2 = new ResponsiveTaskProcessor();
        processor2.start();

        processor2.submitTask(() -> {
            System.out.println("长时间运行任务开始");
            try {
                Thread.sleep(5000);
            } catch (InterruptedException e) {
                System.out.println("长时间运行任务被中断");
                Thread.currentThread().interrupt();
            }
        });

        Thread.sleep(1000);
        processor2.stopImmediately();
    }
}
